package mr

import (
	"fmt"
	"log"
	"net"
	"net/http"
	"net/rpc"
	"os"
	"strconv"
	"sync"
	"time"
)

type Master struct {
	// Your definitions here.
	files   []string
	nReduce int

	splitFile        map[string]IntermediateFileTask
	intermediateFile map[int][]string
	outFile          []string
	currentPhase     int
	undoCounter      int
	doneTaskMap      map[interface{}]*DelayTask

	th *TimeWheel
	mu sync.Mutex
}

type IntermediateFileTask struct {
	filename                  string
	filenameIndex             int
	intermediateFilename      string
	intermediateFilenameIndex int
}

type DelayTask struct {
	id      interface{}
	value   interface{}
	phase   int
	expired time.Time
	done    bool
}

func (t DelayTask) Delay() int64 {
	return (int64)(t.expired.Sub(time.Now()).Seconds())
}

func (t DelayTask) Expired() time.Time {
	return t.expired
}

func (t DelayTask) Done() bool {
	return t.done
}

// Your code here -- RPC handlers for the worker to call.

//
// an example RPC handler.
//
// the RPC argument and reply types are defined in rpc.go.
//
func (m *Master) Example(args *ExampleArgs, reply *ExampleReply) error {
	reply.Y = args.X + 1
	return nil
}

func (m *Master) RequestTask(args *TaskArgs, reply *TaskReply) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	if m.currentPhase == MapPhase && len(m.splitFile) > 0 {
		reply.Phase = MapPhase
		for k, v := range m.splitFile {
			// if _, ok := m.doneTaskMap[k]; ok {
			// 	continue
			// }

			reply.FileIndex = v.intermediateFilenameIndex
			reply.NReduce = m.nReduce
			reply.InputFilename = []string{v.filename}
			reply.OutputFilename = k
			reply.Id = k

			delayTask := DelayTask{
				id:      k,
				value:   v,
				expired: time.Now().Add(time.Duration(10) * time.Second),
				done:    false,
				phase:   MapPhase,
			}
			m.th.AddTask(&delayTask)
			m.doneTaskMap[delayTask.id] = &delayTask
			// fmt.Printf("master 发送 Map 任务: %v\n", k)

			delete(m.splitFile, k)
			break
		}
	} else if m.currentPhase == MapPhase {
		reply.Phase = WaitPhase
	} else if len(m.intermediateFile) > 0 {
		reply.Phase = ReducePhase
		for k, v := range m.intermediateFile {
			// if _, ok := m.doneTaskMap[k]; ok {
			// 	continue
			// }
			reply.NReduce = m.nReduce
			reply.InputFilename = v
			reply.OutputFilename = "mr-out-" + strconv.Itoa(k)
			reply.Id = k

			delayTask := DelayTask{
				id:      k,
				value:   v,
				expired: time.Now().Add(time.Duration(10) * time.Second),
				done:    false,
				phase:   ReducePhase,
			}
			m.th.AddTask(&delayTask)
			m.doneTaskMap[delayTask.id] = &delayTask
			// fmt.Printf("master 发送 Reduce 任务: %v\n", k)

			delete(m.intermediateFile, k)
			break
		}
	} else if m.currentPhase == ReducePhase {
		reply.Phase = WaitPhase
	} else {
		reply.Phase = ExitPhase
	}

	return nil
}

func (m *Master) ReportTask(args *TaskArgs, reply *TaskReply) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	if args.Phase == MapPhase && m.currentPhase == MapPhase {
		id := args.Id

		fmt.Printf("【master】 完成 Map[undo=%d] id=%v, address=%p\n", m.undoCounter, id, m.doneTaskMap[id])

		if _, ok := m.doneTaskMap[id]; !ok || m.doneTaskMap[id].done {
			return nil
		}

		m.undoCounter--

		// fmt.Printf("id=%v\n", id)

		m.doneTaskMap[id].done = true
		if m.undoCounter == 0 {
			m.undoCounter = len(m.intermediateFile)
			m.currentPhase = ReducePhase
		}

		// fmt.Printf("MapPhase: undoCounter=%v\n", m.undoCounter)
	} else if args.Phase == ReducePhase && m.currentPhase == ReducePhase {
		id := args.Id
		fmt.Printf("【master】 完成 Reduce[undo=%d] id=%v, address=%p\n", m.undoCounter, id, m.doneTaskMap[id])

		if _, ok := m.doneTaskMap[id]; !ok || m.doneTaskMap[id].done {
			return nil
		}

		m.undoCounter--
		// fmt.Printf("id=%v\n", id)
		m.doneTaskMap[id].done = true

		if m.undoCounter == 0 {
			m.currentPhase = ExitPhase
		}
		// fmt.Printf("ReducePhase: undoCounter=%v\n", m.undoCounter)
	}

	return nil
}

func (m *Master) handleExpiredTask(tasks []DelayQueueInterface) {

	m.mu.Lock()
	defer m.mu.Unlock()
	for _, v := range tasks {
		if v.Done() == true {
			continue
		}
		task := v.(*DelayTask)
		delete(m.doneTaskMap, task.id)

		fmt.Printf("【检测到未完成的超时任务】id: %v\n", task.id)
		if task.phase == MapPhase {
			id := task.id.(string)
			m.splitFile[id] = (task.value).(IntermediateFileTask)
		} else if task.phase == ReducePhase {
			id := task.id.(int)
			m.intermediateFile[id] = (task.value).([]string)
		}

	}

}

//
// start a thread that listens for RPCs from worker.go
//
func (m *Master) server() {
	rpc.Register(m)
	rpc.HandleHTTP()
	//l, e := net.Listen("tcp", ":1234")
	sockname := masterSock()
	os.Remove(sockname)
	l, e := net.Listen("unix", sockname)
	if e != nil {
		log.Fatal("listen error:", e)
	}
	go http.Serve(l, nil)
}

//
// main/mrmaster.go calls Done() periodically to find out
// if the entire job has finished.
//
func (m *Master) Done() bool {
	ret := false

	// Your code here.
	if m.currentPhase == ExitPhase {
		ret = true
	}

	return ret
}

//
// create a Master.
// main/mrmaster.go calls this function.
// nReduce is the number of reduce tasks to use.
//
func MakeMaster(files []string, nReduce int) *Master {
	m := Master{}

	// Your code here.
	m.mu = sync.Mutex{}
	m.files = files
	m.nReduce = nReduce

	// m.splitFile = make(map[string]int)
	m.splitFile = make(map[string]IntermediateFileTask)
	m.intermediateFile = make(map[int][]string)
	m.outFile = make([]string, 0)

	for i, f := range files {
		for j := 0; j < nReduce; j++ {
			// filename := "inter-" + strconv.Itoa(j) + "-" + f
			filename := f + "-inter-" + strconv.Itoa(j)
			m.splitFile[filename] = IntermediateFileTask{
				filename:                  f,
				filenameIndex:             i,
				intermediateFilename:      filename,
				intermediateFilenameIndex: j,
			}
		}
	}

	for i := 0; i < nReduce; i++ {
		for _, f := range files {
			// filename := "inter-" + strconv.Itoa(i) + "-" + f
			filename := f + "-inter-" + strconv.Itoa(i)
			m.intermediateFile[i] = append(m.intermediateFile[i], filename)
		}
	}

	for i := 0; i < nReduce; i++ {
		m.outFile = append(m.outFile, "mr-out-"+strconv.Itoa(i))
	}

	m.undoCounter = len(m.splitFile)
	m.currentPhase = MapPhase
	m.doneTaskMap = make(map[interface{}]*DelayTask)

	fmt.Printf("splitFile=%v\n", len(m.splitFile))
	fmt.Printf("intermediateFile=%v\n", len(m.intermediateFile))
	fmt.Printf("outFile=%v\n", len(m.outFile))

	m.server()

	m.th = MakeTimeWheel()
	m.th.SetMaster(&m)

	// go m.handleExpiredTask()

	// t1 := Task{
	// 	id:      "t1",
	// 	expired: time.Now().Add(time.Duration(time.Second) * 5),
	// }
	// t2 := Task{
	// 	id:      "t2",
	// 	expired: time.Now().Add(time.Duration(time.Second) * 5),
	// }
	// t3 := Task{
	// 	id:      "t3",
	// 	expired: time.Now().Add(time.Duration(time.Second) * 67),
	// }
	// m.th.AddTask(&t1)
	// m.th.AddTask(&t2)
	// m.th.AddTask(&t3)

	// m.th.PrintEntrys()

	myMap := make(map[string]*DelayTask)
	myMap["shit"] = &DelayTask{}
	delete(myMap, "shit")
	fmt.Printf("%v\n", myMap["shit"])
	if _, ok := myMap["shit"]; ok {
		fmt.Println("???")
	} else {
		fmt.Println("yes")
	}

	return &m
}
